package org.fastsyncer.manager.listener.mysql;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.NullArgumentException;
import org.apache.commons.lang.StringUtils;
import org.fastsyncer.common.constant.ConnectorConstant;
import org.fastsyncer.common.entity.DatabaseConfig;
import org.fastsyncer.common.entity.Mapping;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.manager.listener.AbstractListener;
import org.fastsyncer.manager.listener.mysql.binlog.BinlogEventListener;
import org.fastsyncer.manager.listener.mysql.binlog.BinlogEventV4;
import org.fastsyncer.manager.listener.mysql.binlog.BinlogEventV4Header;
import org.fastsyncer.manager.listener.mysql.binlog.BinlogParser;
import org.fastsyncer.manager.listener.mysql.binlog.BinlogParserContext;
import org.fastsyncer.manager.listener.mysql.binlog.BinlogParserListener;
import org.fastsyncer.manager.listener.mysql.binlog.BinlogRemoteClient;
import org.fastsyncer.manager.listener.mysql.binlog.BinlogRowEventFilter;
import org.fastsyncer.manager.listener.mysql.binlog.impl.event.DeleteRowsEventV2;
import org.fastsyncer.manager.listener.mysql.binlog.impl.event.RotateEvent;
import org.fastsyncer.manager.listener.mysql.binlog.impl.event.TableMapEvent;
import org.fastsyncer.manager.listener.mysql.binlog.impl.event.UpdateRowsEventV2;
import org.fastsyncer.manager.listener.mysql.binlog.impl.event.WriteRowsEventV2;
import org.fastsyncer.manager.listener.mysql.common.glossary.Column;
import org.fastsyncer.manager.listener.mysql.common.glossary.Pair;
import org.fastsyncer.manager.listener.mysql.common.glossary.Row;
import org.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * <h3>监听mysql增量数据</h3>
 * <ol type="1">
 * <li>
 * <dt>实时监听</dt></li>
 * <dd>基于mysql的Client/Server通信协议,抓取binlog,解析增量数据.</dd>
 * <li>
 * <dt>故障切换</dt></li>
 * <dd>当默认连接的机器发生宕机或断开连接时,会重新连接集群中其它机器,直到任一台连接成功或全部连接失败为止.</dd>
 * </ol>
 * 
 * @ClassName: AbstarctMysqlListener
 * @author: AE86
 * @date: 2018年10月18日 下午4:54:16
 */
public abstract class AbstarctMysqlListener extends AbstractListener {

	private static final Logger logger = LoggerFactory.getLogger(AbstarctMysqlListener.class);

	// 默认连接集群中的第一个节点
	private static final Integer DEF_MASTER = 0;
	
	protected DatabaseConfig config;

	private String database;

	private String table;

	private List<String> nameCols;

	private int colsNum;

	// 正在运行的节点
	protected int master;

	// 集群节点连接地址
	protected String[] cluster;

	private volatile BinlogRemoteClient client;
	
	/**
	 * 故障切换中
	 */
	private volatile boolean switching = false;

	@Override
	public void onClose() {
		if (null != client) {
			try {
				synchronized (client) {
					// 注销监听器
					client.getBinlogParser().removeParserListener();
					// 安全停止
					client.stopQuietly();
					logger.info("Stop listener." + driverId);
				}
			} catch (Exception e) {
				logger.error("Close listener: " + e.getLocalizedMessage());
			}
		}
	}

	@Override
	public void run() {
		try {
			// 1、获取驱动
			MappingTask mapping = super.getMapping();
			if (null == mapping) {
				return;
			}
			Mapping sMapping = mapping.getSourceMapping();
			String connector = sMapping.getConnector();
			// 2、检查驱动连接器类型
			if (!StringUtils.equals(ConnectorConstant.MYSQL_CONNECTOR, connector) && !StringUtils.equals(ConnectorConstant.SQL_MYSQL_CONNECTOR, connector)) {
				return;
			}
			config = (DatabaseConfig) sMapping.getConfig();
			table = getTable();
			// 3、配置连接配置地址信息
			setupServerConfig();
			setupNameCol();
			setupMysqlMeta();
			// 4、启动
			launch();
		} catch (Exception e) {
			logger.error("Can not to start listener \"" + driverId + "\" >> " + e.getLocalizedMessage());
		}
	}

	protected String getTable() {
		return config.getTable();
	}

	protected abstract List<Map<String, Object>> getColNames();
	

	private void setupServerConfig() {
		// jdbc:mysql://127.0.0.1:3306,127.0.0.1:3307/test?seUnicode=true&characterEncoding=UTF8&useSSL=true
		String url = config.getUrl();
		if (StringUtils.isBlank(url)) {
			throw new NullArgumentException("The driver url can't be null.");
		}
		Pattern p = Pattern.compile("//(\\S*)/");
		Matcher matcher = p.matcher(url);
		if (!matcher.find()) {
			throw new NullArgumentException("The driver url can't be parsed.");
		}
		// 获取结果：//127.0.0.1:3306,127.0.0.1:3307/
		// 净化参数：127.0.0.1:3306,127.0.0.1:3307
		String str = matcher.group().replaceAll("/", "");
		// 分割字符：[127.0.0.1:3306, 127.0.0.1:3307]
		cluster = str.split(",");
		// 检查个数
		if (null == cluster || 1 > cluster.length) {
			throw new NullArgumentException("The driver url can't be parsed.");
		}
		// /test?
		int bIndex = url.lastIndexOf("/");
		int eIndex = url.indexOf("?");
		if (-1 != bIndex && -1 != eIndex) {
			database = url.substring(bIndex + 1, eIndex);
		}
		if (StringUtils.isBlank(database)) {
			throw new NullArgumentException("The driver database can't be null.");
		}
	}

	private void setupNameCol() {
		List<Map<String, Object>> fileds = this.getColNames();
		if (null == fileds || fileds.isEmpty()) {
			throw new NullPointerException("The driver col can't be null.");
		}
		int size = fileds.size();
		List<String> list = new ArrayList<String>(size);
		Map<String, Object> col = null;
		Object name = null;
		for (int i = 0; i < size; i++) {
			col = fileds.get(i);
			name = col.get("name");
			if (null != name) {
				list.add(String.valueOf(name));
			}
		}
		if (null == list || list.isEmpty()) {
			throw new NullPointerException("The driver col can't be null.");
		}
		nameCols = list;
		colsNum = list.size();
	}

	private void setupMysqlMeta() {
		MysqlMeta meta = super.getMappingMeta(MysqlMeta.class);
		// 默认选举第一个节点为master
		this.master = null != meta ? meta.getMaster() : DEF_MASTER;
	}

	private void launch() throws Exception {
		// 1、读取连接地址
		// 返回数组：[127.0.0.1, 13306]
		String[] server = cluster[master].split(":");
		if (null == server || 2 != server.length) {
			throw new IllegalArgumentException("The driver host and port can't be parsed.");
		}
		final String ip = server[0];
		final int port = Integer.parseInt(server[1]);
		// 127.0.0.1:3306>test.user.123
		String threadSuffixName = new StringBuilder(ip).append(":").append(port).append(">").append(database).append(".").append(table).append(".").append(driverId).toString();
		client = new BinlogRemoteClient(ip, port, config.getUsername(), config.getPassword(), threadSuffixName);
		// 2、故障切换操作时不需要加载历史配置
		if (!switching) {
			// 正常启动时,判断是否有历史增量配置
			MysqlMeta meta = super.getMappingMeta(MysqlMeta.class);
			if (null != meta) {
				client.setBinlogFileName(meta.getBinlogFileName());
				client.setBinlogPosition(meta.getBinlogPosition());
			}
		}
		// 定义拦截器
		client.setFilter(new BinlogRowEventFilter() {
			@Override
			public boolean accepts(BinlogEventV4Header header, BinlogParserContext context, TableMapEvent e) {
				// 仅监听指定数据库下的表数据
				String dName = e.getDatabaseName().toString();
				String tName = e.getTableName().toString();
				return StringUtils.equalsIgnoreCase(database, dName) && StringUtils.equalsIgnoreCase(table, tName);
			}
		});
		// 注册监听binlog事件
		client.setBinlogEventListener(new BinlogEventListener() {
			@Override
			public void onEvents(BinlogEventV4 event) {
				parseEvent(event);
			}
		});
		// 注册监听parser状态事件
		client.setBinlogParserListener(new BinlogParserListener.Adapter() {
			@Override
			public void onStop(BinlogParser parser) {
				logger.error("Connect " + cluster[master] + " interrupted.");
				recover();
			}
		});
		// 启动
		client.start();
		logger.info("Start listener." + driverId);
	}

	/**
	 * 恢复流程：关闭客户端,启动故障恢复线程,尝试切换连接其它机器,直到任一台连接成功或全部连接失败为止.
	 * <pre>利用空闲的线程异步执行恢复任务 假设集群中有A、B、C三台mysql数据库,算法如下: 
	 * 	1.如果A发生宕机,则连接B、C;
	 * 	2.如果B、C也无法连接,最后重连A;
	 * 	3.如果仍然无法连接A,取消故障恢复任务,结束线程.
	 * </pre>
	 */
	private void recover() {
		if (null != client) {
			try {
				client.stopQuietly();
			} catch (Exception e) {
				logger.error(e.getLocalizedMessage());
			}
		}
		CompletableFuture.supplyAsync(() -> {
			if (null != cluster) {
				// 标记状态为故障切换中
				switching = true;
				int down = master;
				int len = cluster.length;
				for (int i = 0; i < len; i++) {
					// 为了提高切换效率,排除故障,优先考虑其他机器
					if (i == down) {
						continue;
					}
					// 唤醒其他机器
					if (wakeupMaster(i)) {
						return true;
					}
				}
				// 最后的希望
				if (wakeupMaster(down)) {
					return true;
				}
			}
			logger.error("Attempt to switch failed, please check MySQL is running.");
			logger.error("The current driver ID is " + driverId + ".");
			return false;
		});
	}

	private boolean wakeupMaster(int fire) {
		master = fire;
		try {
			launch();
		} catch (Exception e) {
			logger.error("Connect " + cluster[master] + " failed.");
			return false;
		}
		logger.info("Attempt to switch " + cluster[master] + " successfully.");
		return true;
	}

	private JSONArray parseColumn(Row columns) throws Exception {
		// 检查默认字段个数
		List<Column> cols = columns.getColumns();
		if (null == cols || colsNum != cols.size()) {
			throw new Exception("The number of default fields is inconsistent with the number of parsed.");
		}

		// 转换为列名称对应值
		JSONArray row = new JSONArray();
		JSONObject attr = null;
		for (int i = 0; i < colsNum; i++) {
			attr = new JSONObject();
			attr.put("name", nameCols.get(i));
			attr.put("value", cols.get(i).toString());
			row.put(attr);
		}
		return row;
	}

	private void parseEvent(BinlogEventV4 event) {
		// FormatDescriptionEvent
		// TableMapEvent
		// UpdateRowsEventV2 WriteRowsEventV2 DeleteRowsEventV2
		// XidEvent
		// 修改
		if (event instanceof UpdateRowsEventV2) {
			UpdateRowsEventV2 u = (UpdateRowsEventV2) event;
			JSONArray msg = parseUpdate(u);
			this.sendMsg(msg);
		}
		// 新增
		else if (event instanceof WriteRowsEventV2) {
			WriteRowsEventV2 w = (WriteRowsEventV2) event;
			JSONArray msg = parseWrite(w);
			this.sendMsg(msg);
		}
		// 删除
		else if (event instanceof DeleteRowsEventV2) {
			DeleteRowsEventV2 d = (DeleteRowsEventV2) event;
			JSONArray msg = parseDelete(d);
			this.sendMsg(msg);
		}
		// 切换binlog
		else if (event instanceof RotateEvent) {
			// 当写入的日志超过系统max_binlog_size参数时,会触发切换新的二进制log文件事件
			RotateEvent r = (RotateEvent) event;
			String newFile = r.getBinlogFilename();
			if (!client.getBinlogFileName().equalsIgnoreCase(newFile)) {
				this.refresh(newFile, r.getBinlogPosition());
			}
		}
		// 刷新增量点
		BinlogEventV4Header header = event.getHeader();
		Long p = header.getNextPosition();
		this.refresh(null, p);
	}

	private JSONArray parseUpdate(UpdateRowsEventV2 u) {
		try {
			JSONArray msg = new JSONArray();
			JSONObject row = new JSONObject();
			row.put("eventType", ConnectorConstant.OPERTION_UPDATE);
			List<Pair<Row>> rows = u.getRows();
			Pair<Row> pair = rows.get(0);
			Row before = pair.getBefore();
			Row after = pair.getAfter();
			row.put("before", parseColumn(before));
			row.put("after", parseColumn(after));
			msg.put(row);
			return msg;
		} catch (Exception e) {
			logger.error(e.getLocalizedMessage());
		}
		return null;
	}

	private JSONArray parseWrite(WriteRowsEventV2 w) {
		try {
			JSONArray msg = new JSONArray();
			JSONObject row = new JSONObject();
			row.put("eventType", ConnectorConstant.OPERTION_INSERT);
			row.put("before", new JSONArray());
			List<Row> rows = w.getRows();
			Row after = rows.get(0);
			row.put("after", parseColumn(after));
			msg.put(row);
			return msg;
		} catch (Exception e) {
			logger.error(e.getLocalizedMessage());
		}
		return null;
	}

	private JSONArray parseDelete(DeleteRowsEventV2 d) {
		try {
			JSONArray msg = new JSONArray();
			JSONObject row = new JSONObject();
			row.put("eventType", ConnectorConstant.OPERTION_DELETE);
			List<Row> rows = d.getRows();
			Row before = rows.get(0);
			row.put("before", parseColumn(before));
			row.put("after", new JSONArray());
			msg.put(row);
			return msg;
		} catch (Exception e) {
			logger.error(e.getLocalizedMessage());
		}
		return null;
	}

	/**
	 * 刷新增量记录点
	 * 
	 * @param newBinlogFilename
	 *            新的binlog文件名称
	 * @param newBinlogPosition
	 *            新的binlog记录点
	 */
	private void refresh(String newBinlogFilename, Long newBinlogPosition) {
		MysqlMeta meta = null;
		if (switching) {
			synchronized (this) {
				if(switching){
					// 使用新的节点配置
					createMysqlMeta(master);
					switching = false;
				}
			}
			return;
		} else {
			meta = super.getMappingMeta(MysqlMeta.class);
			if (null == meta) {
				createMysqlMeta();
				return;
			}
		}

		boolean refreshFilename = StringUtils.isNotBlank(newBinlogFilename);
		if (refreshFilename) {
			meta.setBinlogFileName(newBinlogFilename);
			client.setBinlogFileName(newBinlogFilename);
		}
		boolean refreshPos = null != newBinlogPosition && 0 != newBinlogPosition;
		if (refreshPos) {
			meta.setBinlogPosition(newBinlogPosition);
			client.setBinlogPosition(newBinlogPosition);
		}
		if (refreshFilename || refreshPos) {
			super.saveMappingMeta(meta);
			if (logger.isDebugEnabled()) {
				logger.debug(meta.toString());
			}
		}
	}

	private void createMysqlMeta() {
		createMysqlMeta(DEF_MASTER);
	}
	
	private void createMysqlMeta(Integer i) {
		MysqlMeta meta = new MysqlMeta().setBinlogFileName(client.getBinlogFileName()).setBinlogPosition(client.getBinlogPosition()).setMaster(i);
		super.saveMappingMeta(meta);
	}

}
